package com.github.hwangrq.myrpc.consumer.client;

import com.github.hwangrq.myrpc.api.zk.ZkConstant;
import com.github.hwangrq.myrpc.consumer.handler.RpcClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.DisposableBean;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**S
 * 客户端
 * 1.连接Netty服务端
 * 2.提供给调用者主动关闭资源的方法
 * 3.提供消息发送的方法
 */
@Slf4j
public class RpcClient {

    private EventLoopGroup group;

    private Channel channel;

    /** 自定义事件处理器 */
    private final RpcClientHandler rpcClientHandler = new RpcClientHandler();

    /** 处理自定义事件的线程池 */
    private static final ExecutorService executorService = Executors.newCachedThreadPool();

    private String ip;

    private int port;

    public RpcClient() throws Exception {
        //从ZK服务器上获取服务端IP端口列表
        ZkClient zkClient = new ZkClient(ZkConstant.DEFAULT_SERVER);
        List<String> serverList =  zkClient.getChildren(ZkConstant.NAMESPACE);

        //根据ZK中存储的服务端节点列表启动RPC服务端连接
        initClient(serverList, zkClient);

        //注册该节点监听，如果下面的子节点列表发生变化，则重新进行服务端连接
        zkClient.subscribeChildChanges(ZkConstant.NAMESPACE, new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                System.out.println("检测到有子节点发生改变，重新获取子节点列表");
                initClient(currentChilds, zkClient);
            }
        });
    }

    public String getIp() {
        return ip;
    }

    public int getPort() {
        return port;
    }

    /**
     * 初始化方法-连接Netty服务端
     */
    public void initClient(List<String> serverList, ZkClient zkClient) throws Exception {
        try {
            group = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(rpcClientHandler);
                        }
                    });
            chooseServer(serverList, zkClient);
            long start = System.currentTimeMillis();
            channel = bootstrap.connect(ip, port).sync().channel();
            long end = System.currentTimeMillis();
            long time = (end - start);

            //如果耗时超过5秒，则认为该节点失效，从ZK节点上删除
            String path = ZkConstant.NAMESPACE + "/" + ip + ":" + port;
            if (time > 5000) {
                zkClient.delete(path);
                System.out.println("【任务delete】path:{}, data:{}, 执行时间：{}"+":"+path+"{time:}"+time+"{now}"+ LocalDateTime.now());

            } else {
                //将该节点数据修改为耗时时间
                zkClient.writeData(path, String.valueOf(time));
                System.out.println("【任务writeData】path:{}, data:{}, 执行时间：{}"+":"+path+"{time:}"+time+"{now}"+ LocalDateTime.now());
            }
        } catch (Exception e) {
            System.out.println(e.getMessage());
            close();
            //抛出异常
            throw e;
        }
    }

    private void chooseServer(List<String> serverList, ZkClient zkClient) {
        //获取节点数据(服务器响应时间)
        int chooseTime = Integer.MAX_VALUE ;
        for (String server : serverList) {
            Object o = zkClient.readData(ZkConstant.NAMESPACE + "/" + server);
            if(o == null) {
                continue;
            }
            int time = Integer.parseInt(o.toString());
            if(time < chooseTime) {
                chooseTime = time;
                ip = server.split(":")[0];
                port = Integer.parseInt(server.split(":")[1]);
            }
        }
        //如果ZK服务器上还未保存服务端的响应时间信息，则随机选择一台服务端进行绑定
        if(chooseTime == 0 || ip == null) {
            //随机选择一台服务器进行绑定
            int i = new Random().nextInt(serverList.size());
            String randomServerStr = serverList.get(i);
            ip = randomServerStr.split(":")[0];
            port = Integer.parseInt(randomServerStr.split(":")[1]);
        }
    }

    /**
     * 提供给调用者主动关闭资源的方法
     */
    public void close() {
        if (channel != null) {
            channel.close();
        }
        if (group != null) {
            group.shutdownGracefully();
        }
       System.out.println("====== RpcClient close ======");
    }

    /**
     * 提供消息发送的方法
     */
    public Object send(String msg) throws ExecutionException, InterruptedException {
        rpcClientHandler.setRequestMsg(msg);
        Future<Object> submit = executorService.submit(rpcClientHandler);
        return submit.get();
    }
}
